package com.jy.datapipeline.export.splitkey.split.impl;

import com.jy.datapipeline.common.constants.ExpConstants;
import com.jy.datapipeline.common.exception.GlobalException;
import com.jy.datapipeline.export.common.constants.SplitKeyExecuteConstant;
import com.jy.datapipeline.export.common.dto.RangeDto;
import com.jy.datapipeline.export.common.enums.impl.ZoomOperationEnums;
import com.jy.datapipeline.export.splitkey.split.Split;
import com.jy.datapipeline.export.splitkey.handler.SplitKeyTaskHandler;
import com.jy.datapipeline.export.splitkey.utils.AssertUtil;
import com.jy.datapipeline.export.splitkey.utils.SplitKeyUtil;
import org.springframework.stereotype.Component;

import java.math.BigInteger;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @类名 AccurateSplit
 * @作者 白条君
 * @创建日期 2023/8/25 14:35
 * @版本 1.0
 */
@Component
public class AccurateSplit implements Split {

    @Override
    public void execute(SplitKeyTaskHandler splitKeyTaskHandler) throws Exception {
        final List<RangeDto> NEW_RANGES = new ArrayList<>(); //存放缩放后的新范围
        final List<BigInteger> RANGE_COUNT_LIST = new ArrayList<>(); //rangeCount统计列表
        final Map<BigInteger, BigInteger> RANGE_RIGHT_MAX_MAP = new HashMap<>(); //rangeCount--->rightMax映射

        Map<String, Object> res = splitKeyTaskHandler.initSplitKeyTask();
        int MinSplitPKType = (int) res.get(SplitKeyExecuteConstant.SPLIT_KEY_TYPE);
        //判断是不是数值类型
        if (!AssertUtil.isNumberType(MinSplitPKType)) {
            throw new GlobalException(ExpConstants.ACCURATE_ONLY_SUPPORT_NUMBER);
        }
        //获取用户需要拆分的片数
        BigInteger pieces = new BigInteger(String.valueOf(res.get(SplitKeyExecuteConstant.PIECES)));
        //获取总条数
        BigInteger totalCount = new BigInteger(String.valueOf(res.get(SplitKeyExecuteConstant.TOTAL_COUNT)));
        //获取分片键范围的最大值和最小值
        Object maxSplitPK = res.get(SplitKeyExecuteConstant.MAX_SPLIT_KEY);
        Object minSplitPK = res.get(SplitKeyExecuteConstant.MIN_SPLIT_KEY);
        BigInteger max = new BigInteger(String.valueOf(maxSplitPK));
        BigInteger min = new BigInteger(String.valueOf(minSplitPK));

        //得到分片键之间的间隔
        BigInteger piecesInterval = SplitKeyUtil.getPiecesInterval(max, min, pieces);

        //初始化分片键的取值范围
        RangeDto initRange = SplitKeyUtil.getInitRange(min, piecesInterval);
        BigInteger rightMax = initRange.getMax();
        BigInteger leftMin = initRange.getMin();

        //获取分片基准值
        BigInteger referenceValue = SplitKeyUtil.getReferenceValue(totalCount, pieces);

        //对每个范围分别执行容错率比较或缩放操作
        int count = 0;
        while (true) {
            count++;
            //如果当前范围最大值等于总的最大值，就不用进行容错率比较或缩放操作，即最后一个分片范围不执行缩放
            if (leftMin.compareTo(max) >= 0 || rightMax.compareTo(max) >= 0) {
                //如果是首次执行，就直接将初始范围添加到NEW_RANGES
                if (count == 1) {
                    NEW_RANGES.add(initRange);
                }
                break;
            }
            //根据分片健的范围获取记录总数
            BigInteger rangeCount = splitKeyTaskHandler.getCount(leftMin, rightMax);
            RANGE_COUNT_LIST.add(rangeCount);
            RANGE_RIGHT_MAX_MAP.put(rangeCount, rightMax);
            //计算是否大于容错率
            boolean isAboveErrorRate = SplitKeyUtil.isAboveErrorRate(rangeCount, referenceValue, SplitKeyExecuteConstant.ERROR_RATE);
            //如果低于容错率就瑞出循环，并添加最小值和最大值的范围
            if (!isAboveErrorRate) {
                NEW_RANGES.add(new RangeDto(leftMin, rightMax));
                break;
            }

            //在有限次数内执行缩放操作，同时返回执行下一轮分片键的范围
            RangeDto rangeDto = this.zoomOperation(rangeCount, referenceValue, rightMax, leftMin, splitKeyTaskHandler,
                    piecesInterval, NEW_RANGES, RANGE_COUNT_LIST, RANGE_RIGHT_MAX_MAP, isAboveErrorRate, max);
            rightMax = rangeDto.getMax();
            leftMin = rangeDto.getMin();
        }

        //执行分片导出操作
        this.export(NEW_RANGES, splitKeyTaskHandler);
    }

    /**
     * 执行8次缩放操作
     * @param rangeCount
     * @param referenceValue
     * @param rightMax
     * @param leftMin
     * @param splitKeyTaskHandler
     * @param piecesInterval
     * @param newRanges
     * @param rangeCountList
     * @param rangeRightMaxMap
     * @param isAboveErrorRate
     */
    private RangeDto zoomOperation(BigInteger rangeCount, BigInteger referenceValue, BigInteger rightMax,
                                   BigInteger leftMin, SplitKeyTaskHandler splitKeyTaskHandler, BigInteger piecesInterval,
                                   List<RangeDto> newRanges, List<BigInteger> rangeCountList, Map<BigInteger, BigInteger> rangeRightMaxMap,
                                   boolean isAboveErrorRate, BigInteger max) {
        int tryCount = 0;
        while (true) {
            //但文件分片数大于基准值进行缩小操作
            ZoomOperationEnums zoomOperation;
            if (rangeCount.compareTo(referenceValue) > 0) {
                zoomOperation = ZoomOperationEnums.REDUCE_OPERATION;
                //但文件分片数小于基准值进行放大操作
            } else if (rangeCount.compareTo(referenceValue) < 0) {
                zoomOperation = ZoomOperationEnums.ENLARGE_OPERATION;
            } else {
                throw new GlobalException(ExpConstants.ZOOM_OPERATION_EXP);
            }
            //进行缩放操作后返回新的最大值
            rightMax = SplitKeyUtil.executeZoomOperation(rightMax, SplitKeyExecuteConstant.SCALE_FACTOR, zoomOperation);
            //根据缩放后的分片健范围获取新的记录总数，然后添加到范围统计列表和range--->rightmax映射
            rangeCount = splitKeyTaskHandler.getCount(leftMin, rightMax);
            rangeCountList.add(rangeCount);
            rangeRightMaxMap.put(rangeCount, rightMax);

            /**
             * 如果rangeCount存在于referenceValue的左边界和右边界，或者达到缩放次数上限
             * 就选取rangeCountList中最接近基准值referenceValue的rangeCount的分片键id作为rightMax
             * 修改isAboveErrorRate的值
             */
            if (SplitKeyUtil.isExistsBetweenNumbers(rangeCountList, referenceValue) || ++tryCount == SplitKeyExecuteConstant.RANGE_EXECUTE_COUNT) {
                rangeCount = SplitKeyUtil.getCloseValue(rangeCountList, referenceValue);
                rightMax = rangeRightMaxMap.get(rangeCount);
                isAboveErrorRate = false;
            }

            /**
             * 清除rangeCount缓存和rangeRightMax缓存
             * 如果低于容错率就返回新的范围
             * 将当前最大值赋值给下一轮范围取值的最小值，再把当前最小值+间隔值重新获取范围最大值
             * 退出缩放操作循环
             */
            if (!isAboveErrorRate) {
                rangeCountList.clear();
                rangeRightMaxMap.clear();
                RangeDto rangeDto = new RangeDto(leftMin, rightMax);
                //当前最大值给到下一轮的最小值，当前最小值+间隔值给到下一轮的最大值
                leftMin = rightMax;
                rightMax = leftMin.add(piecesInterval);
                //如果rightmax大于等于max，就把max作为rightmax
                if (rightMax.compareTo(max) >= 0) {
                    rangeDto.setMax(max);
                }
                newRanges.add(rangeDto);
                break;
            }
        }
        return new RangeDto(leftMin, rightMax);
    }

    /**
     * 执行导出操作
     * @param newRanges
     * @param splitKeyTaskHandler
     */
    private void export(List<RangeDto> newRanges, SplitKeyTaskHandler splitKeyTaskHandler) {
        for (int i = 0; i < newRanges.size(); i++) {
            RangeDto rangeDto = newRanges.get(i);
            String dataFileSuffix = SplitKeyUtil.getDataFileSuffix(i);
            boolean isFinish = SplitKeyUtil.getIsFinish(newRanges, i);
            splitKeyTaskHandler.executeSplitKey(rangeDto.getMin(), rangeDto.getMax(), dataFileSuffix, isFinish);
        }
    }
}
